package com.data.capture.serialize;

import com.data.capture.constant.OperationType;
import com.data.capture.mq.dto.ChangeDataCaptureInfo;
import com.ververica.cdc.debezium.DebeziumDeserializationSchema;
import io.debezium.data.Envelope;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.java.typeutils.GenericTypeInfo;
import org.apache.flink.util.Collector;
import org.apache.kafka.connect.data.Field;
import org.apache.kafka.connect.data.Schema;
import org.apache.kafka.connect.data.Struct;
import org.apache.kafka.connect.source.SourceRecord;

import java.util.List;
import java.util.Objects;


/**
 * 数据变更信息序列化器
 * @author linch
 **/
public class CustomDebeziumDeserializationSchema implements DebeziumDeserializationSchema<ChangeDataCaptureInfo> {

    @Override
    public TypeInformation<ChangeDataCaptureInfo> getProducedType() {
        return GenericTypeInfo.of(ChangeDataCaptureInfo.class);
    }

    @Override
    public void deserialize(SourceRecord record, Collector<ChangeDataCaptureInfo> out) {
        ChangeDataCaptureInfo info = new ChangeDataCaptureInfo();
        // 获取操作表名
        String topic = record.topic();
        String[] topicInfos = topic.split("\\.");
        info.setDatabase(topicInfos[1]);
        info.setTableName(topicInfos[2]);
        Struct struct = (Struct) record.value();
        // 操作类型
        Envelope.Operation operation = Envelope.operationFor(record);
        String type = operation.toString().toUpperCase();
        info.setType(type);
        // 获取before数据
        Struct before = struct.getStruct("before");
        if (Objects.nonNull(before)) {
            Schema schema = before.schema();
            List<Field> fields = schema.fields();
            fields.forEach(field -> info.getBeforeMap().put(field.name().toUpperCase(), before.get(field)));
        }
        // 获取after数据
        Struct after = struct.getStruct("after");
        if (Objects.nonNull(after)) {
            Schema schema = after.schema();
            List<Field> fields = schema.fields();
            fields.forEach(field -> info.getAfterMap().put(field.name().toUpperCase(), after.get(field)));
        }
        // 设置变更字段
        if (OperationType.UPDATE.eq(info.getType())) {
            info.getBeforeMap().forEach((k, v) -> {
                if (!StringUtils.equals(String.valueOf(info.getAfterMap().get(k)), String.valueOf(v))) {
                    info.getChangeFields().add(k);
                }
            });
        }
        // 设置主键
        String primaryKey = ((Struct) record.key()).schema().fields().get(0).name();
        String primaryValue = String.valueOf(((Struct)record.key()).get(primaryKey));
        info.setPrimaryKey(primaryKey);
        info.setPrimaryValue(primaryValue);
        out.collect(info);
    }

}
